/* ************************************************************************ */
/*                                                                          */
/*  haXe Video                                                              */
/*  Copyright (c)2007 Nicolas Cannasse                                      */
/*                                                                          */
/* This library is free software; you can redistribute it and/or            */
/* modify it under the terms of the GNU Lesser General Public               */
/* License as published by the Free Software Foundation; either             */
/* version 2.1 of the License, or (at your option) any later version.	      */
/*                                                                          */
/* This library is distributed in the hope that it will be useful,          */
/* but WITHOUT ANY WARRANTY; without even the implied warranty of           */
/* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU        */
/* Lesser General Public License or the LICENSE file for more details.      */
/*                                                                          */
/* ************************************************************************ */
package hxvid;
import hxvid.events.IEvent;
import hxvid.events.ConnectionEvent;
import hxvid.events.StreamEvent;
import hxvid.events.ServerEvent;
import hxvid.events.SharedObjectEvent;
import format.amf.Value;
import format.flv.Data;
import hxvid.Rtmp;
import hxvid.Commands;
import hxvid.SharedObject;
import hxvid.Server;
import hxvid.Data;
import hxvid.Logger;

typedef CommandInfos = {
	var id : Int;
	var h : RtmpHeader;
	var p : RtmpPacket;
	var c : Client;
}

typedef RtmpMessage = {
	header : RtmpHeader,
	packet : RtmpPacket
}

typedef RtmpStream = 
{
	var name: String;
	var id : Int;
	var channel : Int;
	var audio : Bool;
	var video : Bool;
	var cache : List<{ data : RtmpPacket, time : Int }>;
	var play : {
		var file : String;
		var flv : Dynamic; /* format.flv.Reader; */
		var startTime : Float;
		var curTime : Int;
		var blocked : Null<Float>;
		var paused : Null<Float>;
	};
	var record : {
		var file : String;
		var startTime : Float;
		var flv : format.flv.Writer;
		var pubOpt : String;
		var listeners : List<RtmpStream>;
		var bytes : Int;
		var lastPing : Int;
	};
	var shared : {
		var lock : neko.vm.Mutex;
		var stream : RtmpStream;
		var client : Client;
		var paused : Null<Float>;
	};
	var objects : {
		var lock : neko.vm.Mutex;
	};
}

enum ClientState {
	WaitHandshake;
	WaitHandshakeResponse( hs : haxe.io.Bytes );
	Ready;
	WaitBody( h : RtmpHeader, blen : Int );
}

class Client {

	static var file_security = ~/^[A-Za-z0-9_-][A-Za-z0-9_\/-]*(\.flv)?$/;
	static var globalLock = new neko.vm.Mutex();
	
	public var socket : neko.net.Socket;
	public var server : Server;
	public var id:Int;
	public var ip:String;
	public var name:String;
	public var eventHandlers:Hash<Array<Dynamic>>;
	public var commands : Commands<CommandInfos>;
	public var logger:Logger;
	
	var rtmp : Rtmp;
	var state : ClientState;
	var streams : Array<RtmpStream>;
	var awaitingStreams:Hash<RtmpStream>;
	
	var accepted:Bool;
	var reason:String;

	public function new( serv, s )
	{
		state = WaitHandshake;
		server = serv;
		socket = s;
		ip = s.peer().host.toString();
		name="";
		socket.output.bigEndian = true;
		rtmp = new Rtmp(null,socket.output);
		
		logger = new Logger(neko.Sys.getCwd() + "/"+server.config.logFile);
		logger.id = "[?/" + ip + "]";
		commands = new Commands();
		commands.logger = logger;
		initializeCommands();
		
		awaitingStreams = new Hash();
		eventHandlers = new Hash();
		streams = new Array();
		initSoStream();
		accepted = true;
		reason="";
	}
	function initializeCommands() 
	{
		commands.add2("connect",cmdConnect,T.Object,T.Opt(T.Unk));
		commands.add1("createStream",cmdCreateStream,T.Null);
		//commands.add5("play",cmdPlay,T.Null,T.Unk, T.Opt(T.Unk), T.Opt(T.Unk), T.Opt(T.Unk));
		commands.addCustom("play",cmdPlay,[T.Null,T.Unk, T.Opt(T.Unk), T.Opt(T.Unk), T.Opt(T.Unk)]);
		commands.add2("deleteStream",cmdDeleteStream,T.Null,T.Int);
		commands.add3("publish",cmdPublish,T.Null,T.Unk,T.Opt(T.String));
		commands.add3("pause",cmdPause,T.Null,T.Opt(T.Bool),T.Int);
		commands.add3("pauseRaw",cmdPause,T.Null,T.Opt(T.Bool),T.Int);
		commands.add2("receiveAudio",cmdReceiveAudio,T.Null,T.Bool);
		commands.add2("receiveVideo",cmdReceiveVideo,T.Null,T.Bool);
		commands.add1("closeStream",cmdCloseStream,T.Null);
		commands.add2("seek",cmdSeek,T.Null,T.Int);
	}
	public function addCommands(arr:Array<Dynamic>)
	{
		for(args in arr)
			Reflect.callMethod(commands, Reflect.field(commands, "add"+(args.length-2)), args);
	}

	function addData( h : RtmpHeader, data : haxe.io.Bytes, kind, p ) {
		var s = streams[h.src_dst];
		if( s == null )
			throw "Client:addData Unknown stream "+h.src_dst;
		var r = s.record;
		if( r == null )	
			return;//throw "Publish not done on stream "+h.src_dst; //commented (is/should be handled elsewhere)
		var time = Std.int((neko.Sys.time() - r.startTime) * 1000);
		var chunk = kind(data,time);
		if(r.flv != null)//added: (only write to file when published with pubOpt 'record')
		{
			r.flv.writeChunk(chunk);
		}
		r.bytes += data.length;
		if( r.bytes - r.lastPing > 100000 ) {
			rtmp.send(2,PBytesReaded(r.bytes));
			r.lastPing = r.bytes;
		}
		for( s in r.listeners ) {
			s.shared.lock.acquire();
			if( s.shared.paused == null ) {
				if( s.cache == null ) {
					s.cache = new List();
					server.wakeUp(s.shared.client.socket,0);
				}
				s.cache.add({ data : p, time : time });
			}
			s.shared.lock.release();
		}
	}

	function error( i : CommandInfos, msg : String ) {
		rtmp.send(i.h.channel,PCall("onStatus",0,[
			ANull,
			format.amf.Tools.encode({
				level : "error",
				code : "NetStream.Error",
				details : msg,
			})
		]),null,i.h.src_dst);
		logger.error(msg);
	}

	public function getStream( i : CommandInfos, ?play : Bool ) {
		var s = streams[i.h.src_dst];
		if( s == null || (play && s.play == null) )
			error(i,"Invalid stream id "+i.h.src_dst);
		return s;
	}

	function openFLV( file:String ) : Dynamic //format.flv.Reader 
	{
		if(file.indexOf(".f4v")!=-1 || file.indexOf(".mp4")!=-1)
			return openF4V(file);
		var flv = null;
		try {
			flv = new format.flv.Reader(neko.io.File.read(file,true));
			flv.readHeader();
		} catch( e : Dynamic ) {
			if( flv != null ) {
				flv.close();
				neko.Lib.rethrow("Client:openFLV Corrupted FLV File '"+file+"' ("+Std.string(e)+")");
			}
			throw "Client:openFLV FLV file not found '"+file+"'";
		}
		return flv;
	}
	function openF4V( file ) : format.f4v.Reader {
		var f4v = null;
		try {
			f4v = new format.f4v.Reader(neko.io.File.read(file,true));
			f4v.readHeader(file);
		} catch( e : Dynamic ) {
			if( f4v != null ) {
				f4v.close();
				neko.Lib.rethrow("Client:openF4V Corrupted F4V or MP4 File '"+file+"' ("+Std.string(e)+")");
			}
			throw "Client:openF4V F4V or MP4 file not found '"+file+"'";
		}
		return f4v;
	}
	function cmdConnect(i : CommandInfos, obj:Dynamic, ?arg:Dynamic=null ) 
	{
		var app = server.getApplication(this, obj.app);
		if(app==null)
		{
			rtmp.send(i.h.channel,PCall("_result",i.id,[
			ANull,
			format.amf.Tools.encode({
				level : "error",
				code : "NetConnection.Connect.Rejected",
				description : "The application name specified during connect is invalid, or could not create instance of " + obj.app
				})
			]));
			server.stopClient(this.socket);
			return;
		}
		logger.path = neko.Sys.getCwd() + server.config.appsDir + "/" + obj.app + "/" + "log.txt";
		logger.id = obj.app + "/" + ip;
		//server.dispatchEvent(new ServerEvent(ServerEvent.CLIENT_CONNECT, server, {client:this, obj:obj, arg:arg}));
		app.onConnect(this, obj, arg);
		if(!accepted)
		{
			rtmp.send(i.h.channel,PCall("_result",i.id,[
			ANull,
			format.amf.Tools.encode({
				level : "error",
				code : "NetConnection.Connect.Rejected",
				description : "Access denied."
				})
			]));
			server.stopClient(this.socket);
		}
		else
		{
			rtmp.send(i.h.channel,PCall("_result",i.id,[
				ANull,
				format.amf.Tools.encode({
					level : "status",
					code : "NetConnection.Connect.Success",
					description : "Connection succeeded."
				})
			]));
			this.dispatchEvent(new ConnectionEvent(ConnectionEvent.CONNECTED, this));
		}
	}
	public function cleanup() 
	{
		for( s in streams )
			if( s != null )
				closeStream(s,true);

		if(server!=null)
		{
			var sos = server.getSharedObjects(this);
			if(sos!=null)
				for(o in sos) 
					o.removeListener(this);
		}
				
		streams = new Array();
		initSoStream();
		dispatchEvent(new ConnectionEvent(ConnectionEvent.DISCONNECTED, this));
		removeListeners();
	}
	function cmdCreateStream(i : CommandInfos, _ : Dynamic ) 
	{
		var s = allocStream();
		var event = new StreamEvent(StreamEvent.STREAM_CREATE, this, {info:i, id:s.id, name:s.name});
		dispatchEvent(event);
		rtmp.send(i.h.channel,PCall("_result",i.id,[ANull,	ANumber(s.id)]));
	}
	function cmdPlay(i : CommandInfos, file:Dynamic, ?start:Dynamic=null, ?len:Dynamic=null, ?obj:Dynamic=null) 
	{
		var s = streams[i.h.src_dst];
		if( s == null )
		{
			error(i,"Unknown 'play' channel");
		}
		s.name = Std.string(file);
		var event = new StreamEvent(StreamEvent.STREAM_PLAY, this, {info:i, id:s.id, name:s.name, start:start, len:len, obj:obj});
		dispatchEvent(event);
		var sharedStreams = server.getSharedStreams(this);
		if(Std.is(file, Bool) && file==false)
		{
			if( s.play != null )
			{
				awaitingStreams.remove(s.play.file);
				s.play.flv=null;
				seek(s,0);
				sendStatus(s,"NetStream.Play.Stop",{ 
				details : "playback stopped" 
				});
				s.play=null;
			}
			else if(s.shared!=null)
			{
				for(i in sharedStreams)
				{
					for (j in i.record.listeners)
					{
						if(j==s)
						{
							i.record.listeners.remove(s);
						}
					}
				}
				sendStatus(s,"NetStream.Play.Stop",{ 
					details : "playback stopped" 
				});
				s.shared=null;
			}
			return;
		}
		var file = securize(i,file);
		if(file==null)
			return;
			
		var appsDir = server.config.appsDir;
		var appName = server.getApplicationName(this);
		var config = server.getApplicationConfig(this);
		var streamsDir = config.streamsDir;
		var flvPath = appsDir+"/"+appName+"/"+streamsDir+"/"+file;
		if( s.play != null )
		{
			s.play.flv.close();
			sendStatus(s,"NetStream.Play.Stop",{ 
				details : "playback stopped" 
			});
			s.play=null;
		}
		if( s.shared != null )
		{
			var sh = sharedStreams.get(file);
			if(sh!=null)
			{
				sh.record.listeners.remove(s);
				sendStatus(s,"NetStream.Play.Stop",{ 
					details : "playback stopped" 
				});
			}
			s.shared=null;
		}
		s.channel = i.h.channel;

		if(sharedStreams.exists(file))
		{
			globalLock.acquire();
			var sh = sharedStreams.get(file);
			s.shared = {
				lock : new neko.vm.Mutex(),
				client : this, 
				stream : sh, 
				paused : null, 
			};
			sh.record.listeners.add(s);
			globalLock.release();
			seek(s,0);
			sendStatus(s,"NetStream.Play.Reset",{
				description : "Resetting "+file+".", 
				details : file, 
				clientId : s.id
			});
			sendStatus(s,"NetStream.Play.Start",{
				description : "Start playing "+file+".", 
				clientId : s.id 
			});
		} 
		else if(neko.FileSystem.exists(flvPath))
		{
			s.play = {
				file : flvPath, 
				flv : null, 
				startTime : null, 
				curTime : 0, 
				blocked : null, 
				paused : null
			};
			seek(s,0);
			sendStatus(s,"NetStream.Play.Reset",{
				description : "Resetting "+file+".", 
				details : file, 
				clientId : s.id
			});
			sendStatus(s,"NetStream.Play.Start",{
				description : "Start playing "+file+".", 
				clientId : s.id 
			});
		}
		else
		{
			globalLock.acquire();
			s.shared = {
				lock : new neko.vm.Mutex(),
				client : this, 
				stream : null, 
				paused : null, 
			};
			awaitingStreams.set(file,s);
			globalLock.release();
			seek(s,0);
			sendStatus(s,"NetStream.Play.Reset",{
				description : "Resetting "+file+".", 
				details : file, 
				clientId : s.id
			});
			sendStatus(s,"NetStream.Play.Start",{
				description : "Start playing "+file+".", 
				clientId : s.id 
			});
			//sendStatus(s,"NetStream.Play.StreamNotFound",{description : "Stream "+streamName+" not found.", clientId : s.id });
		}
	}
	
	function cmdDeleteStream( i : CommandInfos, _ : Dynamic, stream : Int ) {
		var s = streams[stream];
		if( s == null )
			error(i,"Invalid 'deleteStream' streamid");
		closeStream(s, true);
	}
	function securize( i, file : String ):String
	{
		/*
		var config = server.getApplicationConfig(this);
		if( !file_security.match(file) )
		{
			error(i,"Invalid file name "+file);
			return null;
		}
		*/
		if( file.indexOf(".") == -1 )
			file += ".flv";
		return file;
	}
	function cmdPublish(i:CommandInfos, _:Dynamic, file:Dynamic, ?pubOpt:String)
	{	
		if(pubOpt!="record")
			pubOpt="live";
		var s = streams[i.h.src_dst];
		if(s == null)
		{
			error(i,"Invalid 'publish' streamid'");
			return;
		}
		s.name = Std.string(file);
		var event = new StreamEvent(StreamEvent.STREAM_PUBLISH, this, {info:i, id:s.id, name:s.name, mode:pubOpt});
		dispatchEvent(event);
		//------------------------------------------------------
		var config = server.getApplicationConfig(this);
		var sharedStreams = server.getSharedStreams(this);
		if(file==false)
		{
			globalLock.acquire();
			if(s.record == null)
			{
				globalLock.release();
				return;
			}
			if(s.record.file != null)
			{
				var streamName = s.record.file;
				var stream = sharedStreams.get(streamName);
				for (i in stream.record.listeners)
				{
					i.shared.client.rtmp.send(s.channel,PCall("onStatus",0,[ANull,format.amf.Tools.encode({level:"status", code:"NetStream.UnPublish.Notify"})]),null,i.id);
					i.shared=null;
				}
				sendStatus(s,"NetStream.UnPublish.Success",{details:s.record.file});
				sendStatus(s,"NetStream.Record.Stop",{details:s.record.file});
				sharedStreams.remove(streamName);
				
				if(s.record.pubOpt=="record")
				{
					s.record.flv.close();
					//s.record.flv=null;
					//seek(s,0);
				}
			}
			globalLock.release();
			return;
		}
		//------------------------------------------------------
		var file = securize(i, Std.string(file));
		if(file == null)
		{
			sendStatus(s,"NetStream.Publish.BadName",{details:file});
			closeStream( s, false );
			return;
		}
		if(sharedStreams.exists(file) || s.record != null)
		{
			sendStatus(s,"NetStream.Publish.BadName",{details:s.record.file});
			closeStream( s, false );
			return;
		}
		//-------------------------------------------
		var appName = server.getApplicationName(this);
		var flvPath = server.config.appsDir+"/"+appName+"/"+config.streamsDir+"/"+file;
		s.channel = i.h.channel;
		if(pubOpt == "record")
		{
			var flv = new format.flv.Writer(neko.io.File.write(flvPath,true));
			flv.writeHeader({ hasAudio : true, hasVideo : true, hasMeta : true });
			s.record = {file:file, startTime:neko.Sys.time(), flv:flv, pubOpt:"record", listeners:new List(), bytes:0,lastPing:0,};
			sendStatus(s,"NetStream.Publish.Start",{ details : file });
			sendStatus(s,"NetStream.Record.Start",{ details : file });
		}
		else
		{
			s.record = {file:file, startTime:neko.Sys.time(), flv:null, pubOpt:"live", listeners:new List(), bytes:0,lastPing:0,};
			sendStatus(s,"NetStream.Publish.Start",{ details : file });
		}
		globalLock.acquire();
		sharedStreams.set(file,s);
		handleWaitingStreams(file,s);
		globalLock.release();
	}

	function cmdPause( i : CommandInfos, _ : Dynamic, ?pause : Bool, time : Int ) 
	{
		var s = getStream(i);
		var event = new StreamEvent(StreamEvent.STREAM_PAUSE, this, {info:i, id:s.id, name:s.name, pause:pause, time:time});
		dispatchEvent(event);
		
		
		var p : { paused : Null<Float> } = s.play;
		if( p == null )
			p = s.shared;
		if( p == null )
			return;
		if( pause == null )
			pause = (p.paused == null); // toggle
		if( pause ) {
			if( p.paused == null )
				p.paused = neko.Sys.time();
			rtmp.send(2,PCommand(s.id,CPlay));
		} else {
			if( p.paused != null ) {
				p.paused = null;
				seek(s,time);
			}
		}
		rtmp.send(i.h.channel,PCall("_result",i.id,[
			ANull,
			format.amf.Tools.encode({
				level : "status",
				code : if( pause ) "NetStream.Pause.Notify" else "NetStream.Unpause.Notify",
			})
		]));
	}

	function cmdReceiveAudio( i : CommandInfos, _ : Dynamic, flag : Bool ) {
		var s = getStream(i);
		s.audio = flag;
	}

	function cmdReceiveVideo( i : CommandInfos, _ : Dynamic, flag : Bool ) {
		var s = getStream(i);
		s.video = flag;
	}

	function cmdCloseStream( i : CommandInfos, _ : Dynamic ) 
	{
		var s = getStream(i);
		var event = new StreamEvent(StreamEvent.STREAM_CLOSE, this, {info:i, id:s.id, name:s.name});
		dispatchEvent(event);
		closeStream(s,false);
	}

	function cmdSeek( i : CommandInfos, _ : Dynamic, time : Int ) 
	{
		var s = getStream(i,true);
		var event = new StreamEvent(StreamEvent.STREAM_SEEK, this, {info:i, id:s.id, name:s.name, time:time});
		dispatchEvent(event);
		seek(s,time);
		rtmp.send(s.channel,PCall("_result",0,[
			ANull,
			format.amf.Tools.encode({
				level : "status",
				code : "NetStream.Seek.Notify",
			})
		]),null,s.id);
		sendStatus(s,"NetStream.Play.Start",{
			time : time
		});
	}
	
	public function processPacket( h : RtmpHeader, p : RtmpPacket ) 
	{
		//trace(Type.enumConstructor(p));
		switch( p )
		{
			case PCall(cmd,iid,args):
				var infos = {
					c : this,
					id : iid,
					h : h,
					p : p,
				};
				if(!commands.has(cmd))
				{
					rtmp.send(h.channel,PCall("_error",iid,[
					ANull,
					format.amf.Tools.encode({
						level : "error",
						code : "NetConnection.Call.Failed",
						description : "The NetConnection.call method was not able to invoke the server-side method or command. Unknown command "+cmd+"("+args.join(",")+")"
						})
					]));
				}
				else if(!commands.execute(cmd,infos,args))
				{
					rtmp.send(h.channel,PCall("_error",iid,[
					ANull,
					format.amf.Tools.encode({
						level : "error",
						code : "NetConnection.Call.Failed",
						/*description : "The NetConnection.call method was not able to invoke the server-side method or command. Mismatch arguments for "+cmd+"("+args.join(",")+")"*/
						description : "The NetConnection.call method was not able to invoke the server-side method or command. " + commands.lastError
						})
					]));
				}
			case PAudio(data):
				addData(h,data,FLVAudio,p);
			case PVideo(data):
				addData(h,data,FLVVideo,p);
			case PMeta(data):
				addData(h,data,FLVMeta,p);
			case PCommand(sid,cmd):
				logger.info("PCOMMAND cmd:"+Std.string(cmd)+":stream id:"+sid);//neko.Lib.println("PCOMMAND "+Std.string(cmd)+":"+sid);
			case PBytesReaded(b):
				//neko.Lib.println("BYTESREADED "+b);
			case PShared(so):
				//logger.info("PShared so:"+Std.string(so));//neko.Lib.println("SHARED OBJECT "+Std.string(so));
				processSoPacket(h,p,so);
			case PUnknown(k,data):
				logger.warning("PUnknown k:"+k+" ["+data.length+"bytes]");//neko.Lib.println("UNKNOWN "+k+" ["+data.length+"bytes]");
		}
	}

	function allocStream() {
		var ids = new Array();
		for( s in streams )
		{
			if( s != null )
			{
				ids[s.id] = true;
			}
		}
		var id = 1;
		while( id < ids.length ) 
		{
			if( ids[id] == null )
			{
				break;
			}
			id++;
		}
		var s = {
			id : id,
			name:"",
			channel : null,
			play : null,
			record : null,
			audio : true,
			video : true,
			shared : null,
			cache : null,
			objects : null
		};
		streams[s.id] = s;
		return s;
	}

	function closeStream( s : RtmpStream, ?delete:Bool )
	{
		if( s.play != null && s.play.flv != null )
		{
			s.play.flv.close();
			sendStatus(s,"NetStream.Play.Stop",{ details : "playback stopped" });
		}
		if( s.record != null ) 
		{
			globalLock.acquire();
			var sharedStreams = server.getSharedStreams(this);
			if(sharedStreams==null)
				return;
			var stream = sharedStreams.get(s.record.file); //application.sharedStreams.get(s.record.file);
			for (i in stream.record.listeners)
			{
				i.shared.client.rtmp.send(s.channel,PCall("onStatus",0,[ANull,format.amf.Tools.encode({level:"status", code:"NetStream.UnPublish.Notify"})]),null,i.id);
			}
			sendStatus(s,"NetStream.UnPublish.Success",{details:s.record.file});
			if(s.record.pubOpt=="record")
			{
				s.record.flv.close();
				sendStatus(s,"NetStream.Record.Stop",{details:s.record.file});
			}
			server.getSharedStreams(this).remove(s.record.file);
			globalLock.release();
		}
		if( s.shared != null ) 
		{
			globalLock.acquire();
			// on more check in case our shared stream just closed
			if( s.shared != null )
				if(s.shared.stream!=null)
					if(s.shared.stream.record!=null)
						try
							s.shared.stream.record.listeners.remove(s)
						catch(e:Dynamic)
							logger.error(e);		
			sendStatus(s,"NetStream.Play.Stop",{ details : "playback stopped" });
			globalLock.release();
		}		
		if(delete)
		{
			streams[s.id] = null;
		}
		else
		{
			s.play=null;
			s.record=null;
			s.shared=null;
		}
	}

	function seek( s : RtmpStream, seekTime : Int ) {
		// clear
		rtmp.send(2,PCommand(s.id,CPlay));
		rtmp.send(2,PCommand(s.id,CReset));
		rtmp.send(2,PCommand(s.id,CClear));

		// no need to send more data for shared streams
		if( s.shared != null )
			return;

		// reset infos
		var p = s.play;
		var now = neko.Sys.time();
		p.startTime = now - server.config.flvBufferTime - seekTime / 1000;
		if( p.paused != null )
			p.paused = now;
		p.blocked = null;
		if( p.flv != null )
			p.flv.close();
		p.flv = openFLV(p.file);
		s.cache = new List();

		// prepare to send first audio + video chunk (with null timestamp)
		var audio = s.audio;
		var video = s.video;
		//trace([audio, video]);
		var audioCache = null;
		var metaCache = null;
		while( true ) 
		{
			var c = s.play.flv.readChunk();
			if( c == null )
				break;
			switch( c ) 
			{
				case FLVAudio(data,time):
					//trace('FLVAUDIO time:'+time+",seekTime:" + seekTime + ", data.length:"+data.length);
					if( time < seekTime )
						continue;
					audioCache = { data : PAudio(data), time : time };
					if( !audio )
						break;
					audio = false;
					
				case FLVVideo(data,time):
					var keyframe = format.flv.Tools.isVideoKeyFrame(data);
					//trace('FLVVideo keyframe:'+keyframe+', time:'+time+",seekTime:" + seekTime + ", data.length:"+data.length);
					if( keyframe )
						s.cache = new List();
					if( s.video )
						s.cache.add({ data : PVideo(data), time : time });
					if( time < seekTime )
						continue;
					if( !video )
						break;
					video = false;
					
				case FLVMeta(data,time):
					if( time < seekTime )
						continue;
					if( metaCache != null )
						s.cache.add(metaCache);
					metaCache = { data : PMeta(data), time : time };
					if( seekTime != 0 ) 
					{
						s.cache.add(metaCache);
						metaCache = null;
					}
			}
			if( !audio && !video )
				break;
		}
		if( s.audio && audioCache != null )
			s.cache.push(audioCache);
		if( seekTime == 0 && metaCache != null )
			s.cache.push(metaCache);
	}

	function playShared( s : RtmpStream ) {
		s.shared.lock.acquire();
		try {
			if( s.cache != null )
				while( true ) {
					var f = s.cache.pop();
					if( f == null ) {
						s.cache = null;
						break;
					}
					rtmp.send(s.channel,f.data,f.time,s.id);
					if( server.isBlocking(socket) )
						break;
				}
		} catch( e : Dynamic ) {
			s.shared.lock.release();
			neko.Lib.rethrow("Client:playShared "+e);
		}
		s.shared.lock.release();
	}

	function playFLV( t : Float, s : RtmpStream ) 
	{
		//trace("playFLV t:"+t+", s.id:" + s.id); 
		var p = s.play;
		if( p.paused != null )
			return;
		if( p.blocked != null ) {
			var delta = t - p.blocked;
			p.startTime += delta;
			p.blocked = null;
		}
		if( s.cache != null ) {
			while( true ) {
				var f = s.cache.pop();
				if( f == null ) {
					s.cache = null;
					break;
				}
				rtmp.send(s.channel,f.data,f.time,s.id);
				p.curTime = f.time;
				if( server.isBlocking(socket) ) {
					p.blocked = t;
					return;
				}
			}
		}
		var reltime = Std.int((t - p.startTime) * 1000);
		while( reltime > p.curTime ) 
		{
			var c = p.flv.readChunk();
			if( c == null ) 
			{
				p.flv.close();
				s.play = null;
/*				// this will abort the video before the end
 				sendStatus(s,"NetStream.Play.Stop",{ details : p.file });
				rtmp.send(2,PCommand(s.id,CClear));
				rtmp.send(2,PCommand(s.id,CReset));
*/				return;
			}
			//trace(Type.enumConstructor(c));
			switch( c ) 
			{
				case FLVAudio(data,time):
					//trace('FLVAUDIO time:'+time+", data.length:"+data.length);
					if( s.audio )
					{
						//trace('FLVAUDIO time:'+time+", data.length:"+data.length);
						rtmp.send(s.channel,PAudio(data),time,s.id);
					}
					p.curTime = time;
					
				case FLVVideo(data,time):
					//trace('FLVVideo time:'+time+", data.length:"+data.length);
					if( s.video )
					{
						//trace('FLVVideo time:'+time+", data.length:"+data.length);
						rtmp.send(s.channel,PVideo(data),time,s.id);
					}
					p.curTime = time;
					
				case FLVMeta(data,time):
					rtmp.send(s.channel,PMeta(data),time,s.id);
					p.curTime = time;
			}
			if( server.isBlocking(socket) ) 
			{
				p.blocked = t;
				return;
			}
		}
		server.wakeUp( socket, server.config.flvBufferTime / 2 );
	}

	public function updateTime( t : Float ) 
	{
		for( s in streams )
		{
			if( s != null ) 
			{
				if( s.play != null )
				{
					playFLV(t,s);
				}
				else if(s.shared != null )
				{
					playShared(s);
				}
				else if( s.objects != null) 
				{
					playObjects(s);
				}
			}
		}
	}

	
	function handleWaitingStreams(file,stream):Void
	{
		var clients = server.getClients(this);
		for(client in clients)
		{
			if(client.awaitingStreams.exists(file))
			{
				var sh = client.awaitingStreams.get(file);
				if(sh.shared == null)
					return;
				sh.shared.stream = stream;
				stream.record.listeners.add(sh);
			}
		}
	}
	public function readProgressive( buf, pos, len ) {
		//trace(Type.enumConstructor(state));
		switch( state ) {
		case WaitHandshake:
			if( len < Rtmp.HANDSHAKE_SIZE + 1 )
				return null;
			rtmp.i = new haxe.io.BytesInput(buf,pos,len);
			rtmp.i.bigEndian = true;
			rtmp.readWelcome();
			var hs = rtmp.readHandshake();
			rtmp.writeWelcome();
			rtmp.writeHandshake(hs);
			state = WaitHandshakeResponse(hs);
			return { msg : null, bytes : Rtmp.HANDSHAKE_SIZE + 1 };
		case WaitHandshakeResponse(hs):
			if( len < Rtmp.HANDSHAKE_SIZE )
				return null;
			rtmp.i = new haxe.io.BytesInput(buf,pos,len);
			rtmp.i.bigEndian = true;
			var hs2 = rtmp.readHandshake();
			if( hs.compare(hs2) != 0 )
				throw "Client:readProgressive Invalid Handshake";
			rtmp.writeHandshake(hs);
			state = Ready;
			return { msg : null, bytes : Rtmp.HANDSHAKE_SIZE };
		case Ready:
			var hsize = rtmp.getHeaderSize(buf.get(pos));
			if( len < hsize )
				return null;
			rtmp.i = new haxe.io.BytesInput(buf,pos,len);
			rtmp.i.bigEndian = true;
			var h = rtmp.readHeader();
			state = WaitBody(h,rtmp.bodyLength(h,true));
			return { msg : null, bytes : hsize };
		case WaitBody(h,blen):
			if( len < blen )
				return null;
			rtmp.i = new haxe.io.BytesInput(buf,pos,len);
			rtmp.i.bigEndian = true;
			var p = rtmp.readPacket(h);
			var msg = if( p != null ) { header : h, packet : p } else null;
			state = Ready;
			return { msg : msg, bytes : blen };
		}
		return null;
	}
	
	
	function processSoPacket( h : RtmpHeader, p : RtmpPacket, so : SOData ):Void
	{
		var SOSetAttList:Array<Hash<Dynamic>> = new Array();
		var sharedObjects = server.getSharedObjects(this);
		var appsDir = server.config.appsDir;
		var appName = server.getApplicationName(this);
		var soDir = server.getApplicationConfig(this).soDir;
		for(cmd in so.commands) 
		{	
			switch(cmd) 
			{
				case SOConnect:
					if(so.name == null)
					{
						continue;
					}
					var nameparts = so.name.split("?");
					if(nameparts[0].length == 0)
					{
						continue;
					}
					
					var name = nameparts[0];
					var event = new SharedObjectEvent(SharedObjectEvent.CONNECT, this, {so:so});
					dispatchEvent(event);
					subscribeSharedObject(name, so.persist);
					
				case SODisconnect:
					var event = new SharedObjectEvent(SharedObjectEvent.DISCONNECT, this, {so:so});
					dispatchEvent(event);
					var obj = sharedObjects.get(so.name);
					obj.removeListener(this);
			
				case SOSetAttribute(key,value):
					var event = new SharedObjectEvent(SharedObjectEvent.CHANGE, this, {so:so, key:key, value:value});
					dispatchEvent(event);
					var obj = sharedObjects.get(so.name);
					if( obj == null ) 
					{
						return;
					}
					
					var SOSetAttCmd : Hash<Dynamic> = new Hash();
					SOSetAttCmd.set("client" , this);
					SOSetAttCmd.set("key" , key);
					SOSetAttCmd.set("value" , value);
					SOSetAttCmd.set("version" , obj.getVersion());
					SOSetAttList.push(SOSetAttCmd);
					
				case SOSendMessage(msg):
					var event = new SharedObjectEvent(SharedObjectEvent.SEND, this, {so:so, msg:msg});
					dispatchEvent(event);
					var obj = sharedObjects.get(so.name);
					obj.broadcast(p);
				
				default:
					neko.Lib.println("Unhandled incoming shared object packet: " + cmd);
			}
		}
		if( SOSetAttList.length > 0 )
		{
			var obj = sharedObjects.get(so.name);
			obj.setAttributes(this,SOSetAttList);
		}
	}
	
	function subscribeSharedObject(name:String, persist: Bool) 
	{
		var so : SharedObject;
		var sharedObjects = server.getSharedObjects(this);
		if(sharedObjects.exists(name)) 
		{
			so = sharedObjects.get(name);
		}
		else 
		{
			so = new SharedObject(name, persist, this);
			sharedObjects.set(name,so);
		}
		var event = new SharedObjectEvent(SharedObjectEvent.SUBSCRIBE, this, {so:so});
		dispatchEvent(event);
		so.addListener(this, persist);
	}

	function unsubSharedObject(path:String, name:String) 
	{
		var sharedObjects = server.getSharedObjects(this);
		if(sharedObjects.exists(name)) {
			sharedObjects.get(name).removeListener(this);
		}
	}
	public function addSharedObjectEvent( p:RtmpPacket ) {
		var s = streams[0];
		if(s == null)
			throw("Client:addSharedObjectEvent Unknown objects channel");
		s.objects.lock.acquire();
		if( s.cache == null ) {
			s.cache = new List();
			server.wakeUp(socket,0);
		}
		s.cache.add({ data : p, time : null });
		s.objects.lock.release();
	}

	function initSoStream() {
		// shared object 'stream'
		var s = 
		{
			id : 0,
			name:"",
			channel : 3,
			play : null,
			record : null,
			audio : false,
			video : false,
			shared : null,
			cache : null,
			objects :
			{
				lock : new neko.vm.Mutex()
			}
		};
		streams[0] = s;
	}	
	
	function playObjects( s : RtmpStream ) {
		s.objects.lock.acquire();
		try {
			if( s.cache != null )
				while( true ) {
					var f = s.cache.pop();
					if( f == null ) {
						s.cache = null;
						break;
					}
					rtmp.send(s.channel,f.data);
					if( server.isBlocking(socket) )
						break;
				}
		}
		catch( e : Dynamic ) {
			s.objects.lock.release();
			neko.Lib.rethrow("Client:playObjects "+e);
		}
		s.objects.lock.release();
	}
	
	public function addEventListener(type:String, func:Dynamic)
	{
		if(eventHandlers.exists(type))
		{
			for(f in eventHandlers.get(type))
				if(f == func)
					return;
			eventHandlers.get(type).push(func);
		}
		else
		{
			eventHandlers.set(type,[func]);
		}
	}
	public function removeEventListener(type:String, func:Dynamic)
	{
		if(eventHandlers.exists(type))
		{
			var handlers = eventHandlers.get(type);
			for(i in 0...handlers.length)
				if(Reflect.compareMethods(handlers[i], func))
				{
					handlers.splice(i, 1);
					break;
				}
		}
	}
	public function removeListeners()
	{
		eventHandlers = new Hash();
	}
	public function removeCommands()
	{
		commands = new Commands();
	}
	private function dispatchEvent(event:IEvent):Void
	{
		if(eventHandlers.exists(event.type))
			for(f in eventHandlers.get(event.type))
				try
				{
					f(event);
				}
				catch(e:Dynamic)
				{
					logger.error(e);
					break;
				}
	}
	public function accept()
	{
		accepted = true;
		reason="";
	}
	function reject(msg:String)
	{
		accepted = false;
		reason = msg;
	}
	public function sendStatus( s : RtmpStream, status : String, infos : Dynamic ) 
	{
		infos.code = status;
		infos.level = "status";
		rtmp.send(s.channel,PCall("onStatus",0,[ANull,format.amf.Tools.encode(infos)]),null,s.id);
	}
	public function sendResult(i, arg:Dynamic):Void
	{
		rtmp.send(i.h.channel,PCall("_result",i.id,[ANull,format.amf.Tools.encode(arg)]));
	}
	public function sendError(i, arg:Dynamic):Void
	{
		rtmp.send(i.h.channel,PCall("_error",i.id,[ANull,format.amf.Tools.encode(arg)]));
	}
	public function send(func:String, arg:Dynamic):Void
	{
		rtmp.send(3,PCall(func,0,[ANull,format.amf.Tools.encode(arg)]));
	}
	public function send0(func:String):Void
	{
		rtmp.send(3,PCall(func,0,[ANull]));
	}
	public function send1(func:String, arg1:Dynamic):Void
	{
		rtmp.send(3,PCall(func,0,[ANull,format.amf.Tools.encode(arg1)]));
	}
	public function send2(func:String, arg1:Dynamic, arg2:Dynamic):Void
	{
		rtmp.send(3,PCall(func,0,[ANull,format.amf.Tools.encode(arg1),format.amf.Tools.encode(arg2)]));
	}
	public function send3(func:String, arg1:Dynamic, arg2:Dynamic,arg3:Dynamic):Void
	{
		rtmp.send(3,PCall(func,0,[ANull,format.amf.Tools.encode(arg1),format.amf.Tools.encode(arg2),format.amf.Tools.encode(arg3)]));
	}
}
